package com.sunday.canal.adapter.mongo.adapter;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.sunday.canal.adapter.mongo.config.CanalProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.stream.Collectors;

/**
 * @Description: canal 适配器
 * @Author: zsy
 * @Create: 2023/4/10 22:56
 */
@Slf4j
@Component
@EnableConfigurationProperties({CanalProperties.class})
public class MongoAdapter implements SmartInitializingSingleton {

    @Autowired
    private CanalProperties properties;

    @Override
    public void afterSingletonsInstantiated() {

        Collection<CanalMongoStarter> canalMongoStarters = properties.getDestinations().stream().map(destination -> {
            // 基于zookeeper动态获取canal server的地址，建立链接，其中一台server发生crash，可以支持failover
            CanalConnector connector = CanalConnectors.newClusterConnector(properties.getZkServers(), destination, properties.getUsername(), properties.getPassword());
            final CanalMongoStarter client = new CanalMongoStarter(destination, connector);
            client.start();
            return client;
        }).toList();

        //Runtime.getRuntime().addShutdownHook(new Thread()）方法是用来捕捉程序退出的时刻，在程序退出的时候处理必要的退出准备，如关闭网路，关闭文件等。
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                log.info("## stop the canal client");
                canalMongoStarters.stream().forEach(client -> client.stop());
            } catch (Throwable e) {
                log.warn("## something goes wrong when stopping canal:", e);
            } finally {
                log.info("## canal client is down.");
            }
        }));

    }

}

